// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#pragma once

#include <pollux/exec/window_build.h>

namespace kumo::pollux::exec {
    /// Unlike PartitionStreamingWindowBuild, RowsStreamingWindowBuild is capable of
    /// processing window functions as rows arrive within a single partition,
    /// without the need to wait for the entirewindow partition to be ready. This
    /// approach can significantly reduce memory usage, especially when a single
    /// partition contains a large amount of data. It is particularly suited for
    /// optimizing rank, dense_rank and row_number functions, as well as aggregate
    /// window functions with a default frame.
    class RowsStreamingWindowBuild : public WindowBuild {
    public:
        RowsStreamingWindowBuild(
            const std::shared_ptr<const core::WindowNode> &windowNode,
            pollux::memory::MemoryPool *pool,
            const common::SpillConfig *spillConfig,
            tsan_atomic<bool> *nonReclaimableSection);

        void addInput(RowVectorPtr input) override;

        void spill() override {
            POLLUX_UNREACHABLE();
        }

        std::optional<common::SpillStats> spilledStats() const override {
            return std::nullopt;
        }

        void noMoreInput() override;

        bool hasNextPartition() override;

        std::shared_ptr<WindowPartition> nextPartition() override;

        bool needsInput() override;

    private:
        // Adds input rows to the current partition, or creates a new partition if it
        // does not exist.
        void addPartitionInputs(bool finished);

        // Invoked before add input to ensure there is an open (in-complete) partition
        // to accept new input. The function creates a new one at the tail of
        // 'windowPartitions_' if it is empty or the last partition is already
        // completed.
        void ensureInputPartition();

        // Sets to true if this window node has range frames.
        const bool hasRangeFrame_;

        // Points to the input rows in the current partition.
        std::vector<char *> inputRows_;

        // Used to compare rows based on partitionKeys.
        char *previousRow_ = nullptr;

        /// The output gets next partition from the head of 'windowPartitions_' and
        /// input adds to the next partition from the tail of 'windowPartitions_'.
        std::deque<std::shared_ptr<WindowPartition> > windowPartitions_;
    };
} // namespace kumo::pollux::exec
